<?php
/**
 * Created by PhpStorm.
 * User: sxg
 * Date: 2023/7/11
 * Time: 8:44
 */

namespace Socket\Ms\Protocols;

class MqttClient implements Protocols
{

    public $control_packet_type = 0;
    public $control_packet_type_name = '';

    public static $control_packets = [
        0b00010000 => "CONNECT_PACKET_TYPE",
        0b00100000 => "CONNACK_PACKET_TYPE",
        0b00110000 => "PUBLISH_PACKET_TYPE",
        0b01000000 => "PUBACK_PACKET_TYPE",
        0b01010000 => "PUBREC_PACKET_TYPE",
        0b01100000 => "PUBREL_PACKET_TYPE",
        0b01110000 => "PUBCOMP_PACKET_TYPE",
        0b10000000 => "SUBSCRIBE_PACKET_TYPE",
        0b10010000 => "SUBACK_PACKET_TYPE",
        0b10100000 => "UNSUBSCRIBE_PACKET_TYPE",
        0b10110000 => "UNSUBACK_PACKET_TYPE",
        0b11000000 => "PINGREQ_PACKET_TYPE",
        0b11010000 => "PINGRESP_PACKET_TYPE",
        0b11100000 => "DISCONNECT_PACKET_TYPE",
    ];

    public $remain_length = 0;
    public $header_len = 0;
    public $fix_header_len = 0;

    public $connectAcknowledgeFlags = 0;
    public $connectReturnCode = 0;

    public $keepAlive = 0;

    //订阅报文标识符
    public $subscribePacketIdentifier = 0;
    //取消订阅报文标识符
    public $unsubscribePacketIdentifier = 0;
    //订阅主题
    public $subscribeTopics = [];
    //SUBACK返回码
    public $subackReturnCode = 0;

    //重发标志
    public $publishPackDup = 0;
    //服务质量等级
    public $publishPackQos = 0;
    //保留位
    public $publishPackRetain = 0;
    //主题名
    public $publishPackTopicName = '';
    //主题名长度
    public $publishPackTopicNameLen = 0;
    //publish报文标识符
    public $publishPacketIdentifierMSB = 0;
    public $publishPacketIdentifierLSB = 0;
    //publish消息内容
    public $publishPacketMessage = '';
    //publish消息长度
    public $publishPacketMessageLen = 0;
    //PUBREL报文标识符
    public $pubrelPacketIdentifierMSB = 0;
    public $pubrelPacketIdentifierLSB = 0;
    //PUBREC报文标识符
    public $pubrecPacketIdentifierMSB = 0;
    public $pubrecPacketIdentifierLSB = 0;

    const CONNECT_PACK_TYPE = 0b00010000;
    const CONNACK_PACK_TYPE = 0b00100000;
    const PUBLISH_PACK_TYPE = 0b00110000;
    const PUBACK_PACK_TYPE = 0b01000000;
    const PUBREC_PACK_TYPE = 0b01010000;
    const PUBREL_PACK_TYPE = 0b01100000;
    const PUBCOMP_PACK_TYPE = 0b01110000;
    const SUBSCRIBE_PACK_TYPE = 0b10000000;
    const SUBACK_PACK_TYPE = 0b10010000;
    const UNSUBSCRIBE_PACK_TYPE = 0b10100000;
    const UNSUBACK_PACK_TYPE = 0b10110000;
    const PINGREQ_PACK_TYPE = 0b11000000;
    const PINGRESP_PACK_TYPE = 0b11010000;
    const DISCONNECT_PACK_TYPE = 0b11100000;

    public function Len($data)
    {
        // TODO: Implement Len() method.
        if (strlen($data) < 2) {
            return false;
        }

        $this->header_len = 2;
        $this->fix_header_len = 2;

        $this->control_packet_type = ord($data[0]) & 0xf0;
        $this->remain_length = ord($data[1]);
        $this->control_packet_type_name = self::$control_packets[$this->control_packet_type];
        $ret = false;

        switch ($this->control_packet_type) {
            case self::CONNACK_PACK_TYPE:
                $ret = $this->decodeConnAckPackLen($data);
                break;
            case self::PUBLISH_PACK_TYPE:
                $ret = $this->decodePublishPackLen($data);
                break;
            case self::PUBACK_PACK_TYPE:
                $ret = $this->decodePubAckPacketLen();
                break;
            case self::PUBREC_PACK_TYPE:
                $ret = $this->decodePubRecPacketLen($data);
                break;
            case self::PUBREL_PACK_TYPE:
                $ret = $this->decodePubRelPacketLen($data);
                break;
            case self::PUBCOMP_PACK_TYPE:
                $this->remain_length = 2;   //这里原来写的是0 导致客户端在服务质量等级为2的条件下给订阅者返回PUBCOMP时出错
                // 接收缓冲区数据没有清除 接收到客户端的PUBREC时没有正常解析数据包
                $this->header_len += 2;
                return true;
            case self::SUBACK_PACK_TYPE:
                //订阅主题
                $ret = $this->decodeSubAckPackLen($data);
                break;
            case self::UNSUBACK_PACK_TYPE:
                $ret = $this->decodeUnsubAckPackLen($data);
                break;
            case self::PINGRESP_PACK_TYPE:
                //服务端必须发送 PINGRESP报文响应客户端的PINGREQ报文
                $this->remain_length = 0;
                $this->header_len += 2;
                $ret = true;
                break;
        }
        return $ret;
    }

    public function calculateRemainLen($data) {
        $remainLen = $firstRemainLenByte= $this->remain_length;
        if ($remainLen >= 128) {
            $this->header_len += 1;
            $secondRemainLenByte = ord($data[2]);
            $remainLen = $secondRemainLenByte * 128 + ($firstRemainLenByte & 0x7f);
        }
        if ($remainLen >= 16384) {
            $this->header_len += 1;
            $thirdRemainLenByte = ord($data[3]);
            $remainLen = ($thirdRemainLenByte * 128 + (ord($data[2]) & 0x7f)) * 128 + ($firstRemainLenByte & 0x7f);
        }
        if ($remainLen >= 209712) {
            $this->header_len += 1;
            $fourthRemainLenByte = ord($data[4]);
            $remainLen = (($fourthRemainLenByte * 128 + (ord($data[3]) & 0x7f)) * 128 + (ord($data[2]) & 0x7f)) * 128 + ($firstRemainLenByte & 0x7f);
        }
        $this->remain_length = $remainLen;
        $this->fix_header_len = $this->header_len;
    }

    public function decodeConnAckPackLen($data): bool
    {
        $this->remain_length = 2;
        if (strlen($data) < $this->header_len + $this->remain_length) {
            return false;
        }

        $data = substr($data, $this->header_len);
        $this->connectAcknowledgeFlags = ord($data[0]);
        $this->connectReturnCode = ord($data[1]);
        if ($this->connectReturnCode != 0x00) {
            return true;
        }
        $this->header_len += 2;
        return true;
    }

    public function decodeSubAckPackLen($data): bool
    {
        $this->calculateRemainLen($data);
        if (strlen($data) < $this->header_len + $this->remain_length) {
            return false;
        }

        $data = substr($data, $this->header_len);
        //解析报文标识符
        $packetIdentifierMSB = ord($data[0]);
        $packetIdentifierLSB = ord($data[1]);
        $this->subscribePacketIdentifier |= $packetIdentifierMSB << 8;
        $this->subscribePacketIdentifier |= $packetIdentifierLSB << 0;
        $this->header_len += 2;
        return true;
    }

    public function decodeUnsubAckPackLen($data): bool
    {
        $this->remain_length = 2;
        if (strlen($data) < $this->header_len + $this->remain_length) {
            return false;
        }

        $data = substr($data, $this->header_len);
        $this->unsubscribePacketIdentifier = unpack('nL', $data)['L'];
        $this->header_len += 2;
        return true;
    }

    public function decodePublishPackLen($data): bool
    {
        $this->calculateRemainLen($data);
        if (strlen($data) < $this->header_len + $this->remain_length) {
            return false;
        }
        //如果DUP标志被设置为0，表示这是客户端或服务端第一次请求发送这个PUBLISH报文。
        //客户端或服务端请求重发一个PUBLISH报文时，必须将DUP标志设置为1
        //如果DUP标志被设置为1，表示这可能是一个早前报文请求的重发。
        $firstByte = ord($data[0]);
        $this->publishPackDup = (($firstByte & 0b00001000) == 0b00001000) ? 1 : 0;
        $this->publishPackQos = ($firstByte & 0b00000110) >> 1;
        //客户端发给服务端的PUBLISH报文的保留标志被设置为1，服务端必须存储这个应用消息和它的服务质量等级，以便它可以被分发给未来的主题名匹配的订阅者
        //【这个就懒得搞了啊 不存储了】
        $this->publishPackRetain = ($firstByte & 0x01) == 1 ? 1 : 0;
        $data = substr($data, $this->header_len);

        //解析主题名相关
        $this->publishPackTopicNameLen = unpack("nL", $data)['L'];
        $this->publishPackTopicName = substr($data, 2, $this->publishPackTopicNameLen);
        $this->header_len += (2 + $this->publishPackTopicNameLen);
        //报文标识符
        //只有当QoS等级是1或2时，报文标识符（Packet Identifier）字段才能出现在PUBLISH报文中
        if ($this->publishPackQos > 0) {
            $data = substr($data, 2 + $this->publishPackTopicNameLen);
            $this->publishPacketIdentifierMSB = ord($data[0]);
            $this->publishPacketIdentifierLSB = ord($data[1]);
            $this->header_len += 2;
        }
        return true;
    }

    public function decodePubAckPacketLen(): bool
    {
        $this->remain_length = 2;
        $this->header_len += 2;
        return true;
    }

    public function decodePubRelPacketLen($data): bool
    {
        $this->remain_length = 2;
        if (strlen($data) < $this->header_len + $this->remain_length) {
            return false;
        }
        $data = substr($data, $this->header_len);
        $this->pubrelPacketIdentifierMSB = ord($data[0]);
        $this->pubrelPacketIdentifierLSB = ord($data[1]);
        $this->header_len += 2;
        return true;
    }

    public function decodePubRecPacketLen($data): bool
    {
        $this->remain_length = 2;
        if (strlen($data) < $this->header_len + $this->remain_length) {
            return false;
        }
        $data = substr($data, $this->header_len);
        $this->pubrecPacketIdentifierMSB = ord($data[0]);
        $this->pubrecPacketIdentifierLSB = ord($data[1]);
        $this->header_len += 2;
        return true;
    }

    public function encode($data = '')
    {
        // TODO: Implement encode() method.
        $frame = '';
        switch ($this->control_packet_type) {
            case self::PUBLISH_PACK_TYPE:
                if ($this->publishPackQos == 1 && $this->subscribeTopics[$this->publishPackTopicName] > 0) {
                    $frame = $this->responsePUBACK();
                } else if ($this->publishPackQos == 2 && $this->subscribeTopics[$this->publishPackTopicName] == 2) {
                    $frame = $this->responsePUBREC();
                }
                break;
            case self::PUBREL_PACK_TYPE:
                $frame = $this->responsePUBCOMP();
                break;
        }
        return [strlen($frame), $frame];
    }

    public function decode($data = '')
    {
        // TODO: Implement decode() method.
        switch ($this->control_packet_type) {
            case self::PUBREL_PACK_TYPE:
            case self::PUBREC_PACK_TYPE:
            case self::PUBACK_PACK_TYPE:
            case self::UNSUBACK_PACK_TYPE:
            case self::PINGRESP_PACK_TYPE:
            case self::PUBCOMP_PACK_TYPE:
            case self::DISCONNECT_PACK_TYPE:
                return '';
            case self::SUBACK_PACK_TYPE:
                $this->decodeSubAckPacket($data);
                return '';
            case self::PUBLISH_PACK_TYPE:
                //有效载荷包含将被发布的应用消息。数据的内容和格式是应用特定的。
                //有效载荷的长度这样计算：用固定报头中的剩余长度字段的值减去可变报头的长度。
                //包含零长度有效载荷的PUBLISH报文是合法的。
                $this->decodePublishPacket($data);
                return $this->publishPacketMessage;
        }
    }

    public function msgLen($data = '')
    {
        // TODO: Implement msgLen() method.
        return $this->fix_header_len + $this->remain_length;
    }


    public function decodeSubAckPacket($data)
    {
        $data = substr($data, $this->header_len);
        $this->subackReturnCode = ord($data[0]);
    }


    public function decodePublishPacket($data)
    {
        $this->publishPacketMessage = substr($data, $this->header_len);
        $this->publishPacketMessageLen = strlen($this->publishPacketMessage);
    }



    public function responsePUBLISHPACK($topic, $message, $Qos = 0): string
    {
        $qos = 0x00;
        $retain = 0x00;
        if ($Qos == 1) {
            $qos = 0b00000010;
        } else if ($Qos == 2) {
            $qos = 0b00000100;
        }

        $frame = chr(self::PUBLISH_PACK_TYPE|$qos|$retain);
        $topicLen = strlen($topic);
        $messageLen = strlen($message);
        $remainLength = 2 + $topicLen + $messageLen;
        //如果服务质量大于0 还有两个字节的报文标识符
        if ($Qos > 0) {
            $remainLength += 2;
        }
        //对剩余长度进行编码
        $this->appendRemainLen($remainLength, $frame);
        //可变报头按顺序包含主题名和报文标识符。
        $frame .= pack("n", $topicLen);
        $frame .= $topic;
        if ($Qos > 0) {
            //Qos质量等级大于0有报文标识符
            $frame .= chr(0x00).chr(0x01);
        }
        //有效载荷
        $frame .= $message;
        return $frame;
    }

    public function responsePUBACK(): string
    {
        return chr(self::PUBACK_PACK_TYPE).chr(0x02).chr($this->publishPacketIdentifierMSB).chr($this->publishPacketIdentifierLSB);
    }

    public function responsePUBREC(): string {
        return chr(self::PUBREC_PACK_TYPE).chr(0x02).chr($this->publishPacketIdentifierMSB).chr($this->publishPacketIdentifierLSB);
    }

    public function responsePUBREL() : string {
        return chr(self::PUBREL_PACK_TYPE|0x02).chr(0x02).chr($this->pubrecPacketIdentifierMSB).chr($this->pubrecPacketIdentifierLSB);
    }

    public function responsePUBCOMP(): string
    {
        return chr(self::PUBCOMP_PACK_TYPE).chr(0x02).chr($this->pubrelPacketIdentifierMSB).chr($this->pubrelPacketIdentifierLSB);
    }


    public function appendRemainLen($remainLength, &$frame)
    {
        if ($remainLength <= 127) {
            $frame .= pack("C", $remainLength);
        } else if ($remainLength <= 16383) {
            //两个字节存储剩余长度
            $firstByte = 0x80 + $remainLength % 128;
            $secondByte = floor($remainLength / 128);
            $frame .= pack("CC", $firstByte, $secondByte);
        } else if ($remainLength <= 2097151) {
            //三个字节存储剩余长度
            $firstByte = 0x80 + $remainLength % 128;
            $secondByte = 0x80 + ($remainLength / 128) % 128;
            $thirdByte = floor($remainLength / 128) / 128;
            $frame .= pack("CCC", $firstByte, $secondByte, $thirdByte);
        } else {
            //四个字节存储剩余长度
            $firstByte = 0x80 + $remainLength % 128;
            $secondByte = 0x80 + ($remainLength / 128) % 128;
            $thirdByte = 0x80 + ($remainLength / 128 / 128) % 128;
            $fourthByte = ($remainLength / 128 / 128) / 128;
            $frame .= pack("CCCC", $firstByte, $secondByte, $thirdByte, $fourthByte);
        }
    }

    public function encodeConnectPack($mqttSettings) : string {
        //-------------------固定报头 还有一个剩余长度需要计算--------------------
        $packetType = self::CONNECT_PACK_TYPE;
        //------------------------------可变报头------------------------------
        $protocolNameMSB = 0x00;
        $protocolNameLSB = 0x04;
        $protocolName = 'MQTT';
        $protocolLevel = 0x04;
        $connectFlag = 0b11000000;
        $keepaliveMSB = isset($mqttSettings['keepalive']) ? ($mqttSettings['keepalive'] >> 8) & 0x0f : 0x00;
        $keepaliveLSB = isset($mqttSettings['keepalive']) ? ($mqttSettings['keepalive'] >> 0) & 0x0f : 0x14;
        $variableHeaderLen = 10;
        //-----------------------------有效载荷-------------------------------
        $payLoadLen = 0;
        $clientIdentifier = $mqttSettings['clientIdentifier'] ?? 'hello, world';
        $clientIdentifierLen = strlen($clientIdentifier);
        $clientIdentifierLenMSB = ($clientIdentifierLen >> 8) & 0x0f;
        $clientIdentifierLenLSB = ($clientIdentifierLen >> 0) & 0x0f;
        $payLoadLen += (2 + $clientIdentifierLen);
        //遗嘱主题和遗嘱消息就不设置了
        //用户名密码
        $userName = $mqttSettings['user_name'] ?? 'sxg';
        $userNameLen = strlen($userName);
        $userNameLenMSB = ($userNameLen >> 8) & 0x0f;
        $userNameLenLSB = ($userNameLen >> 0) & 0x0f;
        $payLoadLen += (2 + $userNameLen);

        $password = $mqttSettings['password'] ?? '123456';
        $passwordLen = strlen($password);
        $passwordLenMSB = ($passwordLen >> 8) & 0x0f;
        $passwordLenLSB = ($passwordLen >> 0) & 0x0f;
        $payLoadLen += (2 + $passwordLen);

        //计算剩余长度  等于可变报+有效载荷的长度
        $remainLen = $variableHeaderLen + $payLoadLen;
        $frame = chr($packetType);
        $this->appendRemainLen($remainLen, $frame);

        //拼接可变报头
        $frame .= chr($protocolNameMSB).chr($protocolNameLSB).$protocolName;
        $frame .= chr($protocolLevel);
        $frame .= chr($connectFlag);
        $frame .= chr($keepaliveMSB);
        $frame .= chr($keepaliveLSB);

        //拼接有效载荷
        $frame .= chr($clientIdentifierLenMSB);
        $frame .= chr($clientIdentifierLenLSB);
        $frame .= $clientIdentifier;
        $frame .= chr($userNameLenMSB);
        $frame .= chr($userNameLenLSB);
        $frame .= $userName;
        $frame .= chr($passwordLenMSB);
        $frame .= chr($passwordLenLSB);
        $frame .= $password;
        return $frame;
    }

    public function requestPINGREQ(): string {
        return chr(self::PINGREQ_PACK_TYPE) . chr(0x00);
    }

    public function subscribe($topic, $Qos) : string {
        if (!in_array($Qos, ['0', '1', '2'])) {
            echo "订阅质量等级不合法\n";
            return '';
        }
        $this->subscribeTopics[$topic] = $Qos;
        $remainLen = 0;
        //-----------------固定报头----------------
        $subscribePack = self::SUBSCRIBE_PACK_TYPE;
        //-----------------可变报头----------------
        $packetIdentifierMSB = 0x00;
        $packetIdentifierLSB = 0x01;
        $remainLen += 2;
        //-----------------有效载荷----------------
        $topicLen = strlen($topic);
        $topicLenMSB = ($topicLen >> 8) & 0x0f;
        $topicLenLSB = ($topicLen >> 0) & 0x0f;
        $remainLen += (2 + $topicLen);
        $lastByte = $Qos;
        $remainLen += 1;
        return chr($subscribePack).chr($remainLen)
            .chr($packetIdentifierMSB).chr($packetIdentifierLSB)
            .chr($topicLenMSB).chr($topicLenLSB)
            .$topic.chr($lastByte);
    }

    public function unsubscribe($topic) : string {
        $firstByte = self::UNSUBSCRIBE_PACK_TYPE|0x02;
        $remainLen = 0;
        $unsubscribeMSB = ($this->subscribePacketIdentifier >> 8) & 0x0f;
        $unsubscribeLSB = ($this->subscribePacketIdentifier >> 0) & 0x0f;
        $remainLen += 2;
        $topicLen = strlen($topic);
        $remainLen += 2 + $topicLen;
        return chr($firstByte).chr($remainLen).chr($unsubscribeMSB).chr($unsubscribeLSB).pack('n', $topicLen).$topic;
    }

}